"""Queues"""

__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'JoinableQueue']

import collections
import heapq
from queue import Full, Empty

from . import events
from . import futures
from . import locks
from .tasks import coroutine


class _Waiter:
    """Internal queue utility class"""
    def __init__(self, timeout, timeout_exc, event_loop, fut):
        """Create a future with a timeout. If timeout is not None, it is an int
        or float, expressed in seconds.  It is always a relative time.

        future.set_exception(timeout_exc) is executed after a timeout.

        Waiters are only ever executed once.
        """
        self._event_loop = event_loop
        self.fut = fut
        if timeout is not None:
            self._event_loop.call_later(
                timeout, self.set_exception, timeout_exc)

    def set_exception(self, exception):
        if self.fut:
            fut, self.fut = self.fut, None
            fut.set_exception(exception)

    def set_result(self, result):
        if self.fut:
            fut, self.fut = self.fut, None
            fut.set_result(result)

    def done(self):
        return not self.fut


class Queue:
    """Create a queue object with a given maximum size.

    If `maxsize` is ``None`` (the default) the queue size is unbounded.

    ``Queue(0)`` is a channel, that is, its :meth:`put` method always blocks
    until the item is delivered. (This is unlike the `standard Queue`_, where 0
    means infinite size).

    Also unlike the `standard Queue`_, you can reliably know this Queue's size
    with :meth:`qsize`, since your single-threaded Tornado application won't be
    interrupted between calling :meth:`qsize` and doing an operation on the
    Queue.

    :Parameters:
      - `maxsize`: Optional size limit (no limit by default).
      - `initial`: Optional sequence of initial items.

    .. _`standard Queue`: http://docs.python.org/library/queue.html#Queue.Queue
    """
    def __init__(self, maxsize=None, initial=None):
        self._event_loop = events.get_event_loop()
        if maxsize is not None and maxsize < 0:
            raise ValueError("maxsize can't be negative")
        self._maxsize = maxsize

        # _Waiters
        self._getters = collections.deque([])
        # Pairs of (item, _Waiter)
        self._putters = collections.deque([])
        self._init(maxsize, initial)

    def _init(self, maxsize, initial):
        self._queue = collections.deque()
        if initial:
            for item in initial:
                self._put(item)

    def _get(self):
        return self._queue.popleft()

    def _put(self, item):
        self._queue.append(item)

    def __repr__(self):
        return '<%s at %s %s>' % (
            type(self).__name__, hex(id(self)), self._format())

    def __str__(self):
        return '<%s %s>' % (type(self).__name__, self._format())

    def _format(self):
        result = 'maxsize=%r' % (self.maxsize, )
        if getattr(self, '_queue', None):
            result += ' _queue=%r' % list(self._queue)
        if self._getters:
            result += ' _getters[%s]' % len(self._getters)
        if self._putters:
            result += ' _putters[%s]' % len(self._putters)
        return result

    def _consume_done_getters(self, waiters):
        # Delete waiters at the head of the get() queue who've timed out
        while waiters and waiters[0].done():
            waiters.popleft()

    def _consume_done_putters(self):
        # Delete waiters at the head of the put() queue who've timed out
        while self._putters and self._putters[0][1].done():
            self._putters.popleft()

    def qsize(self):
        """Number of items in the queue"""
        return len(self._queue)

    def get_maxsize(self):
        """Number of items allowed in the queue.

        Setting maxsize to a larger number unblocks callbacks waiting on
        :meth:`put`. Decreasing maxsize below :meth:`qsize` raises
        :exc:`RuntimeError`:

        # TODO

        >>> from tornado import gen
        >>> q = toro.Queue(0)
        >>> @gen.engine
        ... def f():
        ...     print 'Trying to put...'
        ...     yield gen.Task(q.put, 'a')
        ...     print 'Done putting'
        ...
        >>> f()
        Trying to put...
        >>> q.maxsize = 1
        Done putting
        """
        return self._maxsize

    def set_maxsize(self, maxsize):
        if maxsize < self.qsize():
            raise RuntimeError("%s has %d items, can't set maxsize to %d" % (
                repr(self), self.qsize(), maxsize))

        self._maxsize = maxsize
        while self.qsize() < maxsize and self._putters:
            item, putter = self._putters.popleft()
            if not putter.done():
                self._put(item)
                putter.set_result(True)

    maxsize = property(get_maxsize, set_maxsize)

    def empty(self):
        """Return ``True`` if the _queue is empty, ``False`` otherwise."""
        return not self._queue

    def full(self):
        """Return ``True`` if there are `maxsize` items in the queue.

        .. note:: if the Queue was initialized with `maxsize=None`
          (the default), then :meth:`full` is never ``True``.
        """
        if self.maxsize is None:
            return False
        elif self.maxsize == 0:
            return True
        else:
            return self.qsize() == self.maxsize

    def put(self, item, timeout=None, block=True):
        """Put an item into the queue.

        If you yield from the future and `timeout` is ``None`` (the default),
        wait until a free slot is available before adding `item` and
        resuming. ``True`` is sent into your coroutine.

        If there's a coroutine waiting on :meth:`get`, it receives the item and
        runs **before** the coroutine waiting for :meth:`put` resumes.

        If `timeout` is the number of seconds to wait, your coroutine is sent
        ``False`` if no free slot becomes available before the timeout.

        If `block` is ``False``, this method puts an item on the queue if a
        free slot is immediately available, else raises the ``Queue.Full``
        exception.  `timeout` is ignored.

        Returns a Future if `block` is ``True`` (the default).

        :Parameters:
          - `timeout`: Optional, maximum seconds to wait.
          - `block`: If ``True`` (the default) return a Future to be completed
            when the item has been put.
        """
        # XXX some refactoring may be possible
        self._consume_done_getters(self._getters)
        if self._getters:
            assert not self._queue, (
                "queue non-empty, why are _getters waiting?")

            getter = self._getters.popleft()

            self._put(item)
            getter.set_result(self._get())

            if block:
                fut = futures.Future(event_loop=self._event_loop)
                fut.set_result(True)
                return fut
            else:
                return True

        elif self.maxsize == self.qsize() and block:
            fut = futures.Future(event_loop=self._event_loop)
            waiter = _Waiter(timeout, Full, self._event_loop, fut)
            self._putters.append((item, waiter))
            return fut
        elif self.maxsize == self.qsize():
            raise Full
        else:
            self._put(item)
            if block:
                fut = futures.Future(event_loop=self._event_loop)
                fut.set_result(True)
                return fut
            else:
                return True

    def get(self, timeout=None, block=True):
        """Remove and return an item from the queue.

        If you `block` is ``True`` and `timeout` is ``None`` (the default), the
        callback is passed an item as soon as one is available.

        If `timeout` is the number of seconds to wait, your coroutine is sent
        ``False`` if no free slot becomes available before the timeout.

        If `block` is ``False``, this method returns an item if one is
        immediately available, else raises exception ``Queue.Full``. `timeout`
        is ignored.

        :Parameters:
          - `timeout`: Optional, maximum seconds to wait.
          - `block`: If ``True`` (the default) return a Future to be completed
            with the item once available.
        """
        self._consume_done_putters()
        if self._putters:
            assert self.full(), "queue not full, why are _putters waiting?"
            item, putter = self._putters.popleft()
            self._put(item)
            if block:
                fut = futures.Future()
                fut.set_result(self._get())
                # When a getter runs and frees up a slot so this putter can
                # run, we need to defer the put for an iteration
                # of the loop to ensure that _getters and _putters alternate
                # perfectly. See TestChannel2.test_wait.
                # TODO: rename TestChannel2.test_wait.
                self._event_loop.call_soon(putter.set_result, True)
                return fut
            else:
                self._event_loop.call_soon(putter.set_result, True)
                return self._get()
        elif self.qsize():
            if block:
                fut = futures.Future()
                fut.set_result(self._get())
                return fut
            else:
                return self._get()
        elif block:
            fut = futures.Future()
            self._getters.append(
                _Waiter(timeout, Empty, self._event_loop, fut))
            return fut
        else:
            raise Empty


class PriorityQueue(Queue):
    """A subclass of :class:`Queue` that retrieves entries in priority order
    (lowest first).

    Entries are typically tuples of the form: ``(priority number, data)``.

    :Parameters:
      - `maxsize`: Optional size limit (no limit by default).
      - `initial`: Optional sequence of initial items.
    """
    def _init(self, maxsize, initial):
        self._queue = list(initial or [])

    def _put(self, item, heappush=heapq.heappush):
        heappush(self._queue, item)

    def _get(self, heappop=heapq.heappop):
        return heappop(self._queue)


class LifoQueue(Queue):
    """A subclass of :class:`Queue` that retrieves most recently added entries
    first.

    :Parameters:
      - `maxsize`: Optional size limit (no limit by default).
      - `initial`: Optional sequence of initial items.
    """
    def _init(self, maxsize, initial):
        self._queue = list(initial or [])

    def _put(self, item):
        self._queue.append(item)

    def _get(self):
        return self._queue.pop()


class JoinableQueue(Queue):
    """A subclass of :class:`Queue` that additionally has :meth:`task_done`
    and :meth:`join` methods.

    :Parameters:
      - `maxsize`: Optional size limit (no limit by default).
      - `initial`: Optional sequence of initial items.
    """
    def __init__(self, maxsize=None, initial=None):
        self._unfinished_tasks = 0
        self._finished = locks.EventWaiter()
        self._finished.set()
        super(JoinableQueue, self).__init__(maxsize=maxsize, initial=initial)

    def _format(self):
        result = Queue._format(self)
        if self._unfinished_tasks:
            result += ' tasks=%s' % self._unfinished_tasks
        return result

    def _put(self, item):
        super(JoinableQueue, self)._put(item)
        self._unfinished_tasks += 1
        self._finished.clear()

    def task_done(self):
        """Indicate that a formerly enqueued task is complete. Used by queue
        consumers. For each :meth:`get <Queue.get>` used to fetch a task, a
        subsequent call to :meth:`task_done` tells the queue that the
        processing on the task is complete.

        If a :meth:`join` is currently blocking, it will resume when all items
        have been processed (meaning that a :meth:`task_done` call was received
        for every item that had been :meth:`put <Queue.put>` into the queue).

        Raises ``ValueError`` if called more times than there were items placed
        in the queue.
        """
        if self._unfinished_tasks <= 0:
            raise ValueError('task_done() called too many times')
        self._unfinished_tasks -= 1
        if self._unfinished_tasks == 0:
            self._finished.set()

    @coroutine
    def join(self, timeout=None):
        """Block until all items in the queue have been gotten and processed.

        The count of unfinished tasks goes up whenever an item is added to the
        queue.  The count goes down whenever a consumer thread calls
        :meth:`task_done` to indicate that the item was retrieved and all work
        on it is complete. When the count of unfinished tasks drops to zero,
        :meth:`join` unblocks.

        :Parameters:
          - `callback`: Function taking no arguments.
          - `timeout`: Optional, maximum seconds to wait.
        """
        fut = futures.Future()
        if self._unfinished_tasks == 0:
            # XXX: why must we wait for the next tick?
            self._event_loop.call_soon(fut.set_result, True)
            yield from fut
        else:
            yield from self._finished.wait(timeout=timeout)
